package app

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{KafkaUtils, LocationStrategies, ConsumerStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import java.util
object enrollmentGenderCounts {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("helloStream")
    val ssc = new StreamingContext(conf, Seconds(2))
    ssc.sparkContext.setLogLevel("error")
    ssc.checkpoint("./checkpoint")

    val kfkParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.188.128:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "niit1",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topicName = Array("stuInfo")
    val streamRdd = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topicName, kfkParams)
    )

    // 映射到 (学期, (男计数, 女计数))
      // 映射到 (在籍状态, 性别, 计数)
    val enrollmentGenderCounts = streamRdd.map { record =>
      val value = record.value()
      val fields = value.split("\t")
      val enrollmentStatus = fields(6) // 在籍状态字段假设为第7个字段（索引6）
      val gender = if (fields(2).toInt == 1) "1" else "0" // 性别字段假设为第3个字段（索引2）
      ((enrollmentStatus, gender), 1)
    }.reduceByKeyAndWindow(_ + _, Seconds(2), Seconds(2))

     val property = new util.HashMap[String,Object]()
    property.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.188.128:9092")
    property.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    property.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")


      enrollmentGenderCounts.foreachRDD(
      x=>{
        // 不为空的时候 进行打印
        if(!x.isEmpty()){
         println("===========================")
          x.foreach {
            case ((enrollmentStatus, gender), count) =>
              // 格式化为 "A,1,4" 的形式
              val formattedString = s"$enrollmentStatus,$gender,$count"
              println(formattedString)
              val kfkProducer = new KafkaProducer[String, String](property)
              kfkProducer.send(new ProducerRecord[String, String]("stuInfo2", formattedString))
              kfkProducer.close()
          }
          // 窗口函数
        }
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }

}
